# -*- coding: utf-8 -*-
# =============================================================================
#         Desc: 统计SQL执行频率。
#       Author: GGA
#        Email:
#     HomePage:
#      Version: 1.0.0
#   LastChange: 2020-12-20
#      History:
# =============================================================================
import datetime
import json
import logging
import socket
import time

import fire
import urllib3

from utils.logger_helper import LoggerHelper
from utils.mysql_helper import MySQLClient

logger = logging.getLogger()
default_datetime_format = "%m-%d %H:%M:%S"
http_pool = urllib3.PoolManager(num_pools=3)


def init_logger(with_debug: bool = False):
    if with_debug:
        LoggerHelper.init_logger(logger_level=logging.DEBUG)
    else:
        LoggerHelper.init_logger(logger_level=logging.INFO)
    global logger
    logger = LoggerHelper.get_logger()


class MySQLStatementAnalyzer(object):
    def __init__(self, mysql_host: str = "127.0.0.1", mysql_port: int = 3306,
                 mysql_user: str = "root", mysql_password: str = "",
                 check_interval: int = 1, check_times: int = 3600,
                 sort_item: str = "EXEC_COUNT", sort_size: int = 30,
                 show_digest_text: bool = False,
                 monitor_api: str = ""):
        """
        监控performance_schema.events_statements_summary_by_digest来获取SQL执行频率。
        :param mysql_host: MySQL 实例的 IP 地址
        :param mysql_port: MySQL 实例的端口
        :param mysql_user: MySQL 实例的用户名
        :param mysql_password: MySQL 实例的密码
        :param check_interval: 统计周期，单位：秒
        :param check_times: 统计的次数
        :param sort_item: 排序项，可选值：
                SUM_EXEC_COUNT, SUM_TIMER_WAIT, SUM_LOCK_TIME, SUM_ROWS_AFFECTED, SUM_ROWS_SENT, SUM_ROWS_EXAMINED,
                AVG_EXEC_COUNT, AVG_TIMER_WAIT, AVG_LOCK_TIME, AVG_ROWS_AFFECTED, AVG_ROWS_SENT, AVG_ROWS_EXAMINED
        :param show_digest_text: 是否显示指纹
        :param sort_size: 排序项的数量
        """
        self.mysql_host = str(mysql_host).strip()
        self.mysql_port = mysql_port
        self.mysql_user = mysql_user
        self.mysql_password = mysql_password
        self.check_interval = check_interval
        self.check_times = check_times
        self.base_exec_info = None
        self.mysql_client = self._get_mysql_client()
        self.sort_item = sort_item
        self.sort_size = sort_size
        self.show_digest_text = show_digest_text
        self.monitor_api = str(monitor_api).strip()
        self.report_host = self.get_report_host()

    def get_report_host(self):
        if self.mysql_host == "127.0.0.1" or self.mysql_host.lower() == "localhost":
            return socket.gethostname()
        try:
            host_name = socket.gethostbyaddr(self.mysql_host)
            return host_name
        except:
            return self.mysql_host

    def check_config(self):
        sort_items = [
            "SUM_EXEC_COUNT", "SUM_TIMER_WAIT", "SUM_LOCK_TIME",
            "SUM_ROWS_AFFECTED", "SUM_ROWS_SENT", "SUM_ROWS_EXAMINED",
            "AVG_EXEC_COUNT", "AVG_TIMER_WAIT", "AVG_LOCK_TIME",
            "AVG_ROWS_AFFECTED", "AVG_ROWS_SENT", "AVG_ROWS_EXAMINED"
        ]
        if self.sort_item not in sort_items:
            logger.warning("sort_item参数有误，请检查")
            exit(1)

    def _get_mysql_client(self):
        return MySQLClient(
            mysql_host=self.mysql_host, mysql_port=self.mysql_port,
            mysql_user=self.mysql_user, mysql_password=self.mysql_password,
            database_name="mysql"
        )

    def get_exec_info(self):
        sql_script = """
        select 
        IFNULL(SCHEMA_NAME,'mysql') AS SCHEMA_NAME,
        IFNULL(DIGEST,'') AS DIGEST,
        IFNULL(DIGEST_TEXT,'') AS DIGEST_TEXT,
        COUNT_STAR,SUM_TIMER_WAIT,SUM_LOCK_TIME,
        SUM_ROWS_AFFECTED,SUM_ROWS_SENT,SUM_ROWS_EXAMINED
        from performance_schema.events_statements_summary_by_digest 
        order by COUNT_STAR desc 
        limit 300
        """
        statement_items = self.mysql_client.mysql_query(sql_script=sql_script, return_dict=True)
        check_time = time.time()
        statement_map = dict()
        for statement_item in statement_items:
            statement_key = statement_item["SCHEMA_NAME"] + "::" + statement_item["DIGEST"]
            statement_map[statement_key] = statement_item
        return {
            "check_time": check_time,
            "statement_map": statement_map
        }

    def show_exec_items(self, exec_items):
        check_time_str = datetime.datetime.now().strftime(default_datetime_format)
        exec_items = sorted(exec_items, key=lambda item: item[self.sort_item], reverse=True)[0:self.sort_size]
        exec_info_list = list()
        if self.show_digest_text:
            head_temp = "| {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} " \
                        "| {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<32} | {:<20} | {} "
            exec_info = head_temp.format(
                "CHECK_TIME", "SUM_EXEC_COUNT", "AVG_EXEC_COUNT",
                "SUM_TIMER_WAIT", "AVG_TIMER_WAIT",
                "SUM_LOCK_TIME", "AVG_LOCK_TIME",
                "SUM_EXAMINED_ROWS", "AVG_EXAMINED_ROWS",
                "SUM_AFFECTED_ROWS", "AVG_AFFECTED_ROWS",
                "SUM_SENT_ROWS", "AVG_SENT_ROWS",
                "DIGEST", "SCHEMA_NAME", "DIGEST_TEXT"
            )
            exec_info_list.append(exec_info)
            for exec_item in exec_items:
                exec_info = head_temp.format(
                    check_time_str,
                    exec_item["SUM_EXEC_COUNT"], exec_item["AVG_EXEC_COUNT"],
                    exec_item["SUM_TIMER_WAIT"], exec_item["AVG_TIMER_WAIT"],
                    exec_item["SUM_LOCK_TIME"], exec_item["AVG_LOCK_TIME"],
                    exec_item["SUM_ROWS_EXAMINED"], exec_item["AVG_ROWS_EXAMINED"],
                    exec_item["SUM_ROWS_AFFECTED"], exec_item["AVG_ROWS_AFFECTED"],
                    exec_item["SUM_ROWS_SENT"], exec_item["AVG_ROWS_SENT"],
                    exec_item["DIGEST"], exec_item["SCHEMA_NAME"], exec_item["DIGEST_TEXT"],
                )
                exec_info_list.append(exec_info)
        else:
            head_temp = "| {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<20} " \
                        "| {:<20} | {:<20} | {:<20} | {:<20} | {:<20} | {:<32} | {:<20} | "
            exec_info = head_temp.format(
                "CHECK_TIME", "SUM_EXEC_COUNT", "AVG_EXEC_COUNT",
                "SUM_TIMER_WAIT", "AVG_TIMER_WAIT",
                "SUM_LOCK_TIME", "AVG_LOCK_TIME",
                "SUM_EXAMINED_ROWS", "AVG_EXAMINED_ROWS",
                "SUM_AFFECTED_ROWS", "AVG_AFFECTED_ROWS",
                "SUM_SENT_ROWS", "AVG_SENT_ROWS",
                "DIGEST", "SCHEMA_NAME"
            )
            exec_info_list.append(exec_info)
            for exec_item in exec_items:
                exec_info = head_temp.format(
                    check_time_str,
                    exec_item["SUM_EXEC_COUNT"], exec_item["AVG_EXEC_COUNT"],
                    exec_item["SUM_TIMER_WAIT"], exec_item["AVG_TIMER_WAIT"],
                    exec_item["SUM_LOCK_TIME"], exec_item["AVG_LOCK_TIME"],
                    exec_item["SUM_ROWS_EXAMINED"], exec_item["AVG_ROWS_EXAMINED"],
                    exec_item["SUM_ROWS_AFFECTED"], exec_item["AVG_ROWS_AFFECTED"],
                    exec_item["SUM_ROWS_SENT"], exec_item["AVG_ROWS_SENT"],
                    exec_item["DIGEST"], exec_item["SCHEMA_NAME"]
                )
                exec_info_list.append(exec_info)
        if len(exec_info_list) > 1:
            logger.info("\n" + "\n".join(exec_info_list))
        pass

    def upload_metric_data(self, metric_data):
        headers = {'Content-Type': 'application/json'}
        http_pool.request(method='POST', url=self.monitor_api, body=json.dumps(metric_data), headers=headers)

    def upload_exec_items(self, exec_items):
        if self.monitor_api == "":
            return
        check_time_ts = int(time.time() * 1000)
        metric_name = 'mysql_query_digest'
        sub_metric_name = ["APS_EXEC_COUNT", "APS_ROWS_EXAMINED", "APS_TIMER_WAIT"]
        for sub_metric_name in sub_metric_name:
            sort_items = sorted(exec_items, key=lambda item: item[sub_metric_name], reverse=True)[0:self.sort_size]
            for sort_item in sort_items:
                if sort_item["DIGEST"] == "":
                    continue
                metric_data = {
                    "metric": {
                        "__name__": metric_name,
                        "sub_metric": sub_metric_name,
                        "server_host": str(self.report_host),
                        "server_port": str(self.mysql_port),
                        "query_digest": sort_item["DIGEST"],
                        "query_digest_text": sort_item["DIGEST_TEXT"],
                    },
                    "values": [sort_item[sub_metric_name]],
                    "timestamps": [check_time_ts]
                }
                self.upload_metric_data(metric_data=metric_data)
            pass
        pass

    def check_exec_rate(self, base_exec_info, curr_exec_info):
        metric_items = [
            "TIMER_WAIT", "LOCK_TIME",
            "ROWS_AFFECTED", "ROWS_SENT", "ROWS_EXAMINED"
        ]
        diff_time = curr_exec_info["check_time"] - base_exec_info["check_time"]
        curr_map = curr_exec_info["statement_map"]
        base_map = base_exec_info["statement_map"]
        exec_items = list()
        check_time = datetime.datetime.now()
        for statement_key in curr_map.keys():
            if statement_key in base_map.keys():
                base_item = base_map[statement_key]
                curr_item = curr_map[statement_key]
                exec_count = curr_item["COUNT_STAR"] - base_item["COUNT_STAR"]
                if exec_count <= 0:
                    continue
                exec_item = {
                    "SCHEMA_NAME": base_item["SCHEMA_NAME"],
                    "DIGEST": base_item["DIGEST"],
                    "DIGEST_TEXT": base_item["DIGEST_TEXT"],
                    "SUM_EXEC_COUNT": exec_count,
                    "AVG_EXEC_COUNT": round(exec_count / diff_time, 2),
                    "APS_EXEC_COUNT": round(exec_count / diff_time, 2)
                }
                for metric_item in metric_items:
                    diff_metric_value = (curr_item["SUM_" + metric_item] - base_item["SUM_" + metric_item])
                    if metric_item in ["TIMER_WAIT", "LOCK_TIME"]:
                        exec_item["SUM_" + metric_item] = int(diff_metric_value / 1000000000)
                        exec_item["APS_" + metric_item] = int(diff_metric_value / 1000000000 / diff_time)
                        exec_item["AVG_" + metric_item] = int(diff_metric_value / 1000000000 / exec_count)
                    else:
                        exec_item["SUM_" + metric_item] = diff_metric_value
                        exec_item["APS_" + metric_item] = int(diff_metric_value / diff_time)
                        exec_item["AVG_" + metric_item] = int(diff_metric_value / exec_count)
                exec_items.append(exec_item)
            pass
        self.show_exec_items(exec_items)
        self.upload_exec_items(exec_items)

    def loop_check(self):
        self.check_config()
        self.base_exec_info = self.get_exec_info()
        time.sleep(self.check_interval)
        check_index = 0
        logger.info("开始检查")
        while check_index < self.check_times:
            logger.info("开始第{}次检查".format(check_index))
            curr_exec_info = self.get_exec_info()
            self.check_exec_rate(self.base_exec_info, curr_exec_info)
            self.base_exec_info = curr_exec_info
            time.sleep(self.check_interval)
            check_index += 1
        logger.info("完成检查")


def main(mysql_host: str = "127.0.0.1", mysql_port: int = 3306,
         mysql_user: str = "root", mysql_password: str = "",
         check_interval: int = 1, check_times: int = 3600,
         sort_item: str = "AVG_EXEC_COUNT", sort_size: int = 30,
         show_digest_text: bool = False, monitor_api: str = ""):
    """
    监控performance_schema.events_statements_summary_by_digest来获取SQL执行频率。
    :param mysql_host: MySQL 实例的 IP 地址
    :param mysql_port: MySQL 实例的端口
    :param mysql_user: MySQL 实例的用户名
    :param mysql_password: MySQL 实例的密码
    :param check_interval: 统计周期，单位：秒
    :param check_times: 统计的次数
    :param sort_item: 排序项，可选值：
            SUM_EXEC_COUNT, SUM_TIMER_WAIT, SUM_LOCK_TIME, SUM_ROWS_AFFECTED, SUM_ROWS_SENT, SUM_ROWS_EXAMINED,
            AVG_EXEC_COUNT, AVG_TIMER_WAIT, AVG_LOCK_TIME, AVG_ROWS_AFFECTED, AVG_ROWS_SENT, AVG_ROWS_EXAMINED
    :param sort_size: 排序项的数量
    :param show_digest_text: 是否显示指纹
    :param monitor_api: 监控服务器,如果monitor_api未空则不上传，
            victoria-metrics的地址为：http://xx.xx.xx.xx:8428/api/v1/import
    """
    ms = MySQLStatementAnalyzer(
        mysql_host=mysql_host, mysql_port=mysql_port,
        mysql_user=mysql_user, mysql_password=mysql_password,
        check_interval=check_interval, check_times=check_times,
        sort_item=sort_item, sort_size=sort_size,
        show_digest_text=show_digest_text, monitor_api=monitor_api
    )
    ms.loop_check()


if __name__ == '__main__':
    init_logger()
    fire.Fire(main)
